package com.ndp.fb.mq.service.schedulerservice;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.dubbo.config.annotation.Service;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.CommonConstants;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.fbagent.service.RedisParamsFBService;
import com.ndp.fb.model.fbagent.param.APIPriority;
import com.ndp.fb.mq.api.service.schedulerservice.HttprequestSchedulerService;
import com.ndp.fb.rdb.api.EntrustAccountRdbService;
import com.ndp.fb.rdb.model.EntrustAccount;
import com.ndp.fb.redis.api.RedisClient;
import com.ndp.fb.redis.model.RedisParams;
import com.ndp.fb.util.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * Created by infi on 2015/3/3.
 */
@Service
public class HttprequestSchedulerServiceImpl implements HttprequestSchedulerService {
    private static final String STOP_REQEUST_PREFIX = "error_stop_act_";
    private static final String LIMITED_PREFIX = "limited_act_";
    private static final String SCORE_ACT_PREFIX = "score_act_";
    private static Logger logger = LoggerFactory.getLogger(HttprequestSchedulerServiceImpl.class);
    /**
     * 任务类型：0,1,2
     * <li>0,1 广告发布</li>
     * <li>2 统计数据抓取</li>
     */
//    private static int JOB_INDEX = 0;

    @Reference
    private RedisParamsFBService redisParamsFBService;

    @Reference
    private EntrustAccountRdbService entrustAccountRdbService;

    @Autowired
    private RedisClient redisClient;

    @Override
    public void requestConsumer() {
        try {
        	List<EntrustAccount> allAdAccount = entrustAccountRdbService.findAdAccountListByRoles("ADMIN","GENERAL_USER","REPORTS_ONLY");
        	EntrustAccount e = new EntrustAccount();
        	e.setFbAccountId(CommonConstants.ViIRTUAL_FB_ID);
        	allAdAccount.add(e);
            for (final EntrustAccount entrustAccount : allAdAccount) {
                Long fbAccountId = entrustAccount.getFbAccountId();
                if (redisClient.hasKey(LIMITED_PREFIX + fbAccountId)) {
                    logger.warn("fbActId >>>> " + fbAccountId + " 此账号目前超限，不能发送请求!");
                } else if (redisClient.hasKey(STOP_REQEUST_PREFIX + fbAccountId)) {
                    logger.warn("fbActId >>>> " + fbAccountId + " 此账号目前存在274或者10错误，不能发送请求!");
                } else {
                    List<RedisParams> reqs = getRedisParamses(fbAccountId);
                    if (reqs == null || reqs.isEmpty()) {
                        //logger.info("fbActId >>>> " + fbAccountId + "下没有可消费的请求。");
                    } else {
                        logger.info("fbActId >>>> " + fbAccountId + ",本次消费请求个数：" + reqs.size());
                        handleAccountReqs(reqs);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("SchedulerConsumerImpl->handelRequest->异常", e);
        }
    }


    /**
     * 获取账号的请求
     *
     * @param fbAccountId
     * @return
     */
    private List<RedisParams> getRedisParamses(Long fbAccountId) {
        List<RedisParams> reqs = null;

            reqs = consumeRedisParamRequestList(fbAccountId, 0d, APIPriority.Default.getScore(), ConfigCentre.getInteger(ConfigConst.CONSUMER_SCHEDULER_REQUESTCOUNT));
        return reqs;
    }

    private void handleAccountReqs(final List<RedisParams> reqs) {
        ThreadPool.execute(() -> {
            for (RedisParams req : reqs) {
                try {
                    redisParamsFBService.makeRequestAsyn(req);
                } catch (Exception e) {
                    logger.error("消费请求异常 uuid >>> " + req.getUuid(), e);
                    //可能存在重复消费风险
                    redisClient.zadd(SCORE_ACT_PREFIX + req.getAccountId(), req.getScore(), req);
                }
            }
        }
        );
    }


    /**
     * 消费队列
     *
     * @param c       count
     * @param fbActId
     * @return
     */
    private synchronized List<RedisParams> consumeRedisParamRequestList(Long fbActId, Double scMin, Double scMax, Integer c) {
        List<RedisParams> result = new ArrayList<>();
        String key = SCORE_ACT_PREFIX + fbActId;
        Set<Object> objects = redisClient.zPopByScore(key, scMin, scMax, 0L, c.longValue() + 1);
        Double min = 0D;
        Double max = 0D;
        long x = 0;
        for (Object object : objects) {
            try {
                if (object == null) {
                    continue;
                }
                RedisParams param = null;
                if (object instanceof RedisParams) {
                    param = (RedisParams) object;
                } else {
                    logger.warn("消费请求失败>>>>" + object.getClass().getClass());
                    continue;
                }

                if (min == 0D) {
                    min = param.getScore();
                    max = param.getScore();
                } else {
                    if (param.getScore() < min) {
                        min = param.getScore();
                    }
                    if (max < param.getScore()) {
                        max = param.getScore();
                    }
                }
                result.add(param);
            } catch (Exception e) {
                logger.error("消费请求", e);
            } finally {
                Long aLong = redisClient.removeRangeByScore(SCORE_ACT_PREFIX + fbActId, min, max);
                logger.info("removeRangeByScore>>>>>" + aLong);
                min = 0D;
                max = 0D;
            }
        }
        return result;
    }
}
